package com.snowcattle.game.service.rpc.client;

import com.snowcattle.game.bootstrap.manager.LocalManager;
import com.snowcattle.game.common.config.GameServerDiffConfig;
import com.snowcattle.game.common.config.ZooKeeperConfig;
import com.snowcattle.game.common.constant.GlobalConstants;
import com.snowcattle.game.common.constant.Loggers;
import com.snowcattle.game.common.constant.ServiceName;
import com.snowcattle.game.service.IService;
import com.snowcattle.game.service.config.GameServerConfigService;
import com.snowcattle.game.service.rpc.server.zookeeper.ZooKeeperNodeBoEnum;
import com.snowcattle.game.service.rpc.server.zookeeper.ZooKeeperNodeInfo;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooDefs.Perms;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.slf4j.Logger;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * zookeeper发现服务
 */
@Service
public class ZookeeperRpcServiceDiscovery implements IService {

    private static final Logger logger = Loggers.rpcLogger;

    private CountDownLatch countDownLatch = new CountDownLatch(1);

    private volatile Map<ZooKeeperNodeBoEnum, List<ZooKeeperNodeInfo>> nodeMap = new ConcurrentHashMap<>();

    private Random random = new Random();

    private CuratorFramework curatorFramework;

    public void discovery(ZooKeeperNodeBoEnum zooKeeperNodeBoEnum) {
        if (curatorFramework == null) {
            curatorFramework = creatClient();
        }
        try {
            setListenter(curatorFramework, zooKeeperNodeBoEnum);
        } catch (Exception e) {
            if (logger.isDebugEnabled()) {
                logger.debug("CuratorFramework Listenning Exception:" + e.getMessage());
            }
        }
    }

    private ZooKeeperNodeInfo chooseService(ZooKeeperNodeBoEnum zooKeeperNodeBoEnum) {
        ZooKeeperNodeInfo data = null;
        List<ZooKeeperNodeInfo> nodeList = nodeMap.get(zooKeeperNodeBoEnum);
        if (nodeList != null) {
            int size = nodeList.size();
            if (size > 0) {
                if (size == 1) {
                    data = nodeList.get(0);
                    if (logger.isDebugEnabled()) {
                        logger.debug("use only data: ", data);
                    }
                } else {
                    data = nodeList.get(random.nextInt(size));
                    if (logger.isDebugEnabled()) {
                        logger.debug("use random data:", data);
                    }
                }
            }
        }

        return data;
    }

    /**
     * 链接zookeeper
     *
     * @return
     */
    private ZooKeeper connectServer() {
        ZooKeeper zk = null;
        try {
            GameServerConfigService gameServerConfigService = LocalManager.getInstance().getLocalSpringServiceManager().getGameServerConfigService();
            ZooKeeperConfig zooKeeperConfig = gameServerConfigService.getZooKeeperConfig();
            String registryAdress = zooKeeperConfig.getProperty(GlobalConstants.ZooKeeperConstants.registryAdress);
            zk = new ZooKeeper(registryAdress, GlobalConstants.ZooKeeperConstants.ZK_SESSION_TIMEOUT, event -> countDownLatch.countDown());
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
        return zk;
    }

    public void stop() {
        if (curatorFramework != null) {
            try {
                curatorFramework.close();
            } catch (Exception e) {
                logger.error(e.toString(), e);
            }
        }
    }

    public List<ZooKeeperNodeInfo> getNodeList(final ZooKeeperNodeBoEnum zooKeeperNodeBoEnum) {
        return nodeMap.get(zooKeeperNodeBoEnum);
    }


    @Override
    public String getId() {
        return ServiceName.ZookeeperRpcServiceDiscovery;
    }

    @Override
    public void startup() throws Exception {
        GameServerConfigService gameServerConfigService = LocalManager.getInstance().getLocalSpringServiceManager().getGameServerConfigService();
        GameServerDiffConfig gameServerDiffConfig = gameServerConfigService.getGameServerDiffConfig();
        if (gameServerDiffConfig.isZookeeperFlag()) {
            try {
                curatorFramework = creatClient();
            } catch (Exception e) {
                logger.error("Create CuratorFramework Client Exception:" + e.getMessage());
            }
            ZooKeeperNodeBoEnum[] zooKeeperNodeBoEnums = ZooKeeperNodeBoEnum.values();
            for (ZooKeeperNodeBoEnum zooKeeperNodeBoEnum : zooKeeperNodeBoEnums) {
                discovery(zooKeeperNodeBoEnum);
            }
        }
    }

    @Override
    public void shutdown() throws Exception {
        if (curatorFramework != null) {
            try {
                curatorFramework.close();
            } catch (Exception e) {
                logger.error(e.toString(), e);
            }
        }
    }

    private static CuratorFramework creatClient() {
        ACLProvider aclProvider = new ACLProvider() {
            private List<ACL> acl;

            @Override
            public List<ACL> getDefaultAcl() {
                if (acl == null) {
                    ArrayList<ACL> acl = ZooDefs.Ids.CREATOR_ALL_ACL;
                    acl.clear();
                    acl.add(new ACL(Perms.ALL, new Id("auth", "admin:admin")));
                    this.acl = acl;
                }
                return acl;
            }

            @Override
            public List<ACL> getAclForPath(String path) {
                return acl;
            }
        };
        //String scheme = "digest";
        //byte[] auth = "admin:admin".getBytes();
        int connectionTimeoutMs = 5000;
        //String connectString = "192.168.0.158:2181";
        GameServerConfigService gameServerConfigService = LocalManager.getInstance().getLocalSpringServiceManager().getGameServerConfigService();
        ZooKeeperConfig zooKeeperConfig = gameServerConfigService.getZooKeeperConfig();
        String registryAdress = zooKeeperConfig.getProperty(GlobalConstants.ZooKeeperConstants.registryAdress);
        String namespace = "";
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .aclProvider(aclProvider)//.authorization(scheme, auth)
                .connectionTimeoutMs(connectionTimeoutMs)
                .connectString(registryAdress)//.namespace(namespace)
                .retryPolicy(new RetryNTimes(Integer.MAX_VALUE, 1000)).build();
        client.start();
        return client;
    }

    private void setListenter(CuratorFramework client, ZooKeeperNodeBoEnum zooKeeperNodeBoEnum) throws Exception {
        ExecutorService pool = Executors.newCachedThreadPool();
        List<String> childrens = client.getChildren().forPath(zooKeeperNodeBoEnum.getRootPath());
        List<ZooKeeperNodeInfo> tempNodeList = new ArrayList<>();
        for (String node : childrens) {
            ZooKeeperNodeInfo zooKeeperNodeInfo = new ZooKeeperNodeInfo();
            byte[] bytes = client.getData().forPath(zooKeeperNodeBoEnum.getRootPath() + "/" + node);
            if (bytes != null) {
                zooKeeperNodeInfo.deserialize(new String(bytes));
                tempNodeList.add(zooKeeperNodeInfo);
            }
        }
        if (logger.isDebugEnabled()) {
            logger.debug("node data: {}", tempNodeList);
        }
        nodeMap.put(zooKeeperNodeBoEnum, tempNodeList);
        if (logger.isDebugEnabled()) {
            logger.debug("Service discovery triggered updating connected server node.");
        }

        RpcClientConnectService rpcClientConnectService = LocalManager.getInstance().getLocalSpringServicerAfterManager().getRpcClientConnectService();
        rpcClientConnectService.notifyConnect(zooKeeperNodeBoEnum, nodeMap.get(zooKeeperNodeBoEnum));

        CuratorCache curatorCache = CuratorCache.builder(client, zooKeeperNodeBoEnum.getRootPath()).build();
        curatorCache.listenable().addListener((type, oldData, newData) -> {
            switch (type) {
                case NODE_CHANGED:
                    if (logger.isDebugEnabled()) {
                        logger.debug("NODE_UPDATED : " + oldData.getPath() + "  数据:" + new String(oldData.getData()));
                    }
                    break;
                case NODE_CREATED:
                    if (logger.isDebugEnabled()) {
                        logger.debug("NODE_ADDED : " + newData.getPath() + "  数据:" + new String(newData.getData()));
                    }
                    break;
                case NODE_DELETED:
                    if (logger.isDebugEnabled()) {
                        logger.debug("NODE_REMOVED : " + oldData.getPath() + "  数据:" + new String(oldData.getData()));
                    }
                    break;
            }
        });

        // 开始监听
        curatorCache.start();
    }
}
